package start.redisstream;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.stereotype.Component;
import org.springframework.util.ErrorHandler;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.UUID;
import java.util.function.Predicate;

@Slf4j
@Component
public class RedisListenRunner implements ApplicationRunner {


    @Autowired
    RedisConnectionFactory redisConnectionFactory;

    @Autowired
    private StringRedisTemplate redisTemplate;

    //@Autowired
    //private ProductUpdateStreamListener productUpdateStreamListener;


    @PostConstruct
    public void init() {
        Boolean aBoolean = redisTemplate.hasKey("lzm_stream");
        if (aBoolean) {
            redisTemplate.delete("lzm_stream");
        }
        //组概念意义 在于每个组都能完整的消费一遍steam中所有数据
        redisTemplate.opsForStream().createGroup("lzm_stream", "lzm_group");
        //redisTemplate.opsForStream().createGroup("lzm_stream", "lzm_group02");
        log.info("创建消费者组：[key:{}]-[group:{}]", "lzm_stream", "lzm_group");
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> containerOptions = StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                // 一次最多拉取20条消息
                .batchSize(20)
                // 拉取消息的超时时间
                //.pollTimeout(Duration.ofMillis(100000))
                //.errorHandler(el -> log.info(el.getMessage()))
                .build();

        // 流消息订阅者容器
        StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer =
                StreamMessageListenerContainer.create(redisConnectionFactory,
                containerOptions);
        // 使用消费组,ReadOffset.lastConsumed()在消费组最后消耗的消息ID之后读取。消费组内的消费者名称需在配置文件中配置
        // 需要注意stream和消费组需提前创建好，XGROUP CREATE yyj_stream yyj_group 0-0 MKSTREAM
        // 要在接收时自动确认消息，请使用receiveAutoAck代替receive

        // 经验证一个消费组内的多个消费者名称可以相同，不会重复消费，解决了集群部署不好区别消费者名称的问题
        String uid1 = "consumer";
        StreamMessageListenerContainer.ConsumerStreamReadRequest<String> stringConsumerStreamReadRequest01 = StreamMessageListenerContainer.StreamReadRequest
                .builder(StreamOffset.create("lzm_stream", ReadOffset.lastConsumed()))
                .consumer(Consumer.from("lzm_group", uid1))
                //.cancelOnError(this.errorHandle())
                .build();

        String uid2 = "consumer";
        StreamMessageListenerContainer.ConsumerStreamReadRequest<String> stringConsumerStreamReadRequest02 = StreamMessageListenerContainer.StreamReadRequest
                .builder(StreamOffset.create("lzm_stream", ReadOffset.lastConsumed()))
                .consumer(Consumer.from("lzm_group", uid2))
                .autoAcknowledge(true)
                //.cancelOnError(this.errorHandle())
                .build();

        streamMessageListenerContainer.register(stringConsumerStreamReadRequest01, new ProductUpdateStreamListener(uid1, "lzm_group"));
        streamMessageListenerContainer.register(stringConsumerStreamReadRequest02, new ProductUpdateStreamListener(uid2, "lzm_group"));
        streamMessageListenerContainer.start();

    }

    @SneakyThrows
    public Predicate<Throwable> errorHandle() {
        return (el) -> {
            try {
                log.info("监听出现异常，异常原因[{}],停顿10s", el.getMessage());
                Thread.sleep(10*1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return false;
        };
    }
}
